package com.xuan.temper.controller;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;

/**
 * @创建人： king xuan
 * @创建时间：2020/9/16
 * @描述：
 */
@Component
public class MQProducer {
    private static final Logger LOGGER = LoggerFactory.getLogger(MQProducer.class);

    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;


    private final DefaultMQProducer producer = new DefaultMQProducer("RocketMQPushConsumer");

    /**
     * 初始化
     */
    @PostConstruct
    public void start() {
        try {
            LOGGER.info("MQ：启动生产者");
            // 1.1 设置NameServer连接地址，如果集群环境，多个之间用分号分隔
            producer.setNamesrvAddr(namesrvAddr);
            // 1.2 启动，开始连接
            producer.start();
        } catch (MQClientException e) {
            LOGGER.error("MQ：启动生产者失败：{}-{}", e.getResponseCode(), e.getErrorMessage());
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    /**
     * 发送消息
     *
     * @param data  消息内容
     * @param topic 主题
     * @param tags  标签
     * @param keys  唯一主键
     */
    public void sendMessage(String data, String topic, String tags, String keys) {
        try {
            byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET);

            Message mqMsg = new Message(topic, tags, keys, messageBody);

            // 3. 消息生产者发送消息
            producer.send(mqMsg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list,
                                           Message message, Object i) {
                    return list.get((Integer) i);
                }

            }, 1); // 队列的索引号，从0开始
            //LOGGER.info("MQ: 生产者发送消息 {}", mqMsg);
            LOGGER.info("MQ: 生产者发送消息 ");
            System.out.println("发送成功");
//            producer.send(mqMsg, new SendCallback() {
//                @Override
//                public void onSuccess(SendResult sendResult) {
//                    LOGGER.info("MQ: 生产者发送消息 {}", sendResult);
//                }
//
//                @Override
//                public void onException(Throwable throwable) {
//                    LOGGER.error(throwable.getMessage(), throwable);
//                }
//            });
        } catch (Exception e) {
            System.out.println("发送失败");
            LOGGER.error(e.getMessage(), e);
        }

    }

    @PreDestroy
    public void stop() {
        if (producer != null) {
            producer.shutdown();
            LOGGER.info("MQ：关闭生产者");
        }
    }
}
